package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import com.atguigu.gmall.realtime.utils.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import redis.clients.jedis.Jedis;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author 黄凯
 * @date 2023/7/18
 * 基本维度关联的实现
 * 模板方法设计模式
 *      在父类中定义完成某一个功能的核心算法的骨架(步骤)，但是有些步骤在父类中没有办法去完成，可以延迟到子类中完成，
 *      在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T,T> implements DimJoinFunction<T> {

    private Connection conn;
    private Jedis jedis;
    private String tableName;

    public DimMapFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getConnection();
        jedis = RedisUtil.getRedisClient();
    }


    @Override
    public T map(T obj) throws Exception {

        //根据流中对象获取要关联的维度主键
        String key = getKey(obj);

        //根据维度的主键获取维度对象
        JSONObject dimInfo = DimUtil.getDimInfo(jedis, conn, GmallConfig.HBASE_NAMESPACE, tableName, key);

        //将维度对象的属性补充到流中的对象上
        if (dimInfo != null) {

            join(obj,dimInfo);

        }

        return obj;
    }
}
